/**
 * Copyright (C) 2009+ Spark Zheng
 *
 * @file        test/def_udp_press.c 
 * @date        2010-05-16
 * @brief	test_module
 *		def_udp protocol press-testing
 *		Two Method:
 *		-1. create pretty many threads
 *		-2. create pretty many fds, use event-and-nonblock client
 * @see         dserver
 * @history
 */

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/resource.h>

#include <dlib/log.h>
#include <dlib/str.h>

#include "libdserver.h"

static void _handle_input(int fd, short event, void *arg);
static void _handle_output(int fd, short event, void *arg);
static void *_press_thread(void *arg);
static int press_threadpool(int concurrence);
static int press_eventpool(int concurrence);

struct press_addr {
	const char *ip;
	in_port_t port;
} addr;

int main(int argc, char **argv)
{
	if (argc < 3) {
		fprintf(stderr, "Usage: def_udp_press type concurrence\n");
		fprintf(stderr, "type: 1 - threadpool, 2 - eventpool\n");
		return -1;
	}

	int type = atoi(argv[1]);
	int concurrence = atoi(argv[2]);

	if (argc > 3) {
		addr.ip = argv[3];
	} else {
		addr.ip = "127.0.0.1";
	}
	if (argc > 4) {
		addr.port = (in_port_t)atoi(argv[4]);
	} else {
		addr.port = 10011;
	}

	fprintf(stderr, "PRESS BEGIN:\b");
	if (type == 1) {
		struct rlimit rlimit;

		getrlimit(RLIMIT_NPROC, &rlimit);
		fprintf(stderr, "cur NPROC: %lu, max NPROC: %lu\n",
				rlimit.rlim_cur, rlimit.rlim_max);
		if ((uint32_t)concurrence > rlimit.rlim_cur &&
				(uint32_t)concurrence < rlimit.rlim_max) {
			rlimit.rlim_cur = concurrence + 1;
			setrlimit(RLIMIT_NPROC, &rlimit);
		}

		press_threadpool(concurrence);
	} else {
		press_eventpool(concurrence);
	}
	fprintf(stderr, "PRESS END.\n");

	return 0;
}

static pthread_mutex_t mutex;

static int press_threadpool(int concurrence)
{
	int i;
	pthread_t pid[concurrence];
	pthread_attr_t attr;

	pthread_attr_init(&attr);
	//pthread_attr_setstacksize(&attr, 128); /* 1024 * 1024  * 3 / 128 = 24576 */
	pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
	pthread_mutex_init(&mutex, NULL);

	for (i = 0; i < concurrence; i++) {
		pthread_mutex_lock(&mutex);
		if (pthread_create(&pid[i], &attr, _press_thread, &i) != 0) {
			fprintf(stderr, "create press thread err[%d]:%s\n", i, strerror(errno));
			pthread_mutex_unlock(&mutex);
		} else {
			fprintf(stderr, "create press thread[%d] success.\n", i);
		}
	}

	pause();
	/*
	for (i = 0; i < concurrence; i++) {
		pthread_join(pid[i], NULL);
	}
	*/
	
	return 0;
}

struct client {
	int		sfd;
	struct event	event;
};

static int press_eventpool(int concurrence)
{
	int i;
	struct client client[concurrence];

	event_init();

	for (i = 0; i < concurrence; i++) {
		client[i].sfd = dserver_udpclient_init(addr.ip, addr.port, 30);
		if (client[i].sfd < 0) {
			fprintf(stderr, "create udpclient error.\n");
			continue;
		} else {
			fprintf(stderr, "create sfd = %d.\n", client[i].sfd);
		}

		event_set(&client[i].event, client[i].sfd, EV_WRITE, _handle_output, &client[i]);
		event_add(&client[i].event, NULL);
	}

	fprintf(stderr, "eventpool begin loop.\n");
	event_dispatch();

	return 0;
}

#define TEST_LOOP	10
#define TEST_BUF_LEN	128	

static char *test_str[TEST_LOOP] = {
	"test1",
	"test2",
	"test3",
	"test4",
	"test5",
	"test6",
	"test7",
	"test8",
	"test9",
	"testA"
};

static void *_press_thread(void *arg)
{
	int i;
	int ret = *(int *)arg;
	int sfd;

	fprintf(stderr, "IN thread %d.\n", ret);

	pthread_mutex_unlock(&mutex);

	char buf[TEST_BUF_LEN + 1] = {0};

	while (1) {
		sfd = dserver_udpclient_init(addr.ip, addr.port, 30);
		if (sfd < 0) {
			fprintf(stderr, "create udpclient error.\n");
			return NULL;
		} else {
			fprintf(stderr, "create sfd = %d.\n", sfd);
		}

		for (i = 0; i < TEST_LOOP; i++) {
			fprintf(stderr, "udp write data[%s].\n", test_str[i]);
			ret = dserver_udpdata_write(sfd, test_str[i], strlen(test_str[i]), 1);
			if (ret < 0) {
				continue;	
			}
			ret = dserver_udpdata_read(sfd, buf, TEST_BUF_LEN, 1);
			if (ret < 0) {
				continue;	
			}
			buf[ret] = 0;
			fprintf(stderr, "udp read data[%s].\n", buf);
		}

		dserver_udpclient_exit(sfd);
	//	usleep(500);
	}
}

static void _handle_input(int fd, short event, void *arg)
{
	int ret;
	struct client *client = (struct client *)arg;
	char buf[TEST_BUF_LEN + 1] = {0};

	UNUSED_PARAM(fd);
	UNUSED_PARAM(event);

	ret = read(client->sfd, buf, TEST_BUF_LEN);
	if (ret > 0) {
		fprintf(stderr, "udp read data[%s].\n", buf);
		event_set(&client->event, client->sfd, EV_WRITE, _handle_output, client);
		event_add(&client->event, NULL);
	} else if (ret == 0) {
		fprintf(stderr, "udp read FIN, close.\n");
		dserver_udpclient_exit(client->sfd);
	} else {
		event_set(&client->event, client->sfd, EV_READ, _handle_input, client);
		event_add(&client->event, NULL);
	}

	return;
}

static void _handle_output(int fd, short event, void *arg)
{
	int ret;
	struct client *client = (struct client *)arg;
	int i = random() % 10;

	UNUSED_PARAM(fd);
	UNUSED_PARAM(event);

	fprintf(stderr, "udp write data[%s].\n", test_str[i]);
	ret = write(client->sfd, test_str[i], strlen(test_str[i]));
	if (ret > 0) {
		event_set(&client->event, client->sfd, EV_READ, _handle_input, client);
		event_add(&client->event, NULL);
	} else {
		event_set(&client->event, client->sfd, EV_WRITE, _handle_output, client);
		event_add(&client->event, NULL);
	}

	return;
}
